// postgres_io_binary.cpp : binary input and output from postgreSQL database
//

#define SEQ_BATCH_SIZE 20000

#include <stdio.h>
#include <fstream>
#include <iostream>
#include <assert.h>
#include <algorithm>
#include <deque>
#include <iostream>
#include "components/libpq++/libpq++.h"
#define _MY_STL_MACROS 1
//#include "components/postgres_io_binary.h"
#include "components/stlhelper.h"
#include "components/progress_indicator.h"
#include "components/pg_db_cursor.h"
#include "components/pg_copy_guard.h"
#include "transfer_data_to_postgres.h"
#include "panda.h"

using std::ostream;
using std::string;
using std::sort;
using std::vector;
using std::deque;



namespace
{
	bool pg_print_sql_err(t_pg_cursor& pg_db,
								 const string& table_name,
								 const string& description,
								 const string& sql_cmd,
								 ostream& CERR)
	{
		CERR		<< "\nERROR!! PostgreSQL error.\n\t"
					<< "Could not "
					<< description
					<< " from the " << table_name << " table: \n\t["
					<< sql_cmd << "]\n\t"
					<< pg_db.ErrorMessage();
		return false;
	}
}

//_______________________________________________________________________________________

//		pg_import_PIs_from_table

//			get all the panda numbers for a specified table

//_______________________________________________________________________________________
bool pg_import_PIs_from_table(t_pg_cursor&	pg_db,
							  const string&	table_name,
							  deque<t_pi>&	vec_PIs,
							  unsigned		PI_cnt_estimate,
							  ostream&		VERBOSE,
							  ostream&		CERR)
{

	// create binary cursor
	string sql_cmd = "SELECT pi FROM " + table_name;
	if (!pg_db.Declare(sql_cmd, true))
		return pg_print_sql_err(pg_db, table_name,
					"create a binary cursor to retrieve PIs", sql_cmd, CERR);

	unsigned batch_size = PI_cnt_estimate ? (PI_cnt_estimate / 50 + 1) : SEQ_BATCH_SIZE;
	// get data in batches
	t_progress_indicator dots(VERBOSE, 1);
	while (pg_db.Fetch(batch_size))
	{
		int curr_batch_size = pg_db.Tuples();
		if (!curr_batch_size)
			break;
		dots++;
		for (int i = 0; i < curr_batch_size; ++i)
			vec_PIs.push_back(t_pi(pg_db.getvalue<unsigned>(i, 0)));

	}

	// close cursor
	if (!pg_db.Close())
		return pg_print_sql_err(pg_db, table_name, "CLOSE CURSOR",
					"close a binary cursor for retrieving PIs", CERR);

	// get rid of excess memory
	sort(beg_to_end(vec_PIs));
	remove_duplicates(vec_PIs);
	deque<t_pi>(vec_PIs).swap(vec_PIs);
	return true;

}

//____________________________________________________________________________________

//	pg_export_PIs

//		export PIs to a specified table
//____________________________________________________________________________________
bool pg_export_PIs_to_table(	t_pg_cursor& pg_db,
								const string& table_name,
								const deque<t_pi>& vec_PIs,
								ostream& VERBOSE, ostream& CERR)
{
	if (!vec_PIs.size())
		return true;

	t_pg_copy_guard copy_buffer(pg_db, table_name, CERR);
	if (!copy_buffer)
		return false;
	

	// loop through each sequence	
	t_progress_indicator dots(VERBOSE, vec_PIs.size() / 50 + 1);
	
	for(unsigned i = 0; i < vec_PIs.size(); ++i, ++dots)
	{
		ultoa(vec_PIs[i], copy_buffer.buf) += "\n";
		copy_buffer.cached_write();
	}

	if (!copy_buffer.end_copy())
		return false;
	return true;
}




bool pg_remove_PIs_from_table(	t_pg_cursor& pg_db,
	 							const string& table_name,
	 							const deque<t_id>& vec_PIs,
								ostream& VERBOSE,
								ostream& CERR,
								string archived_columns)
{
	if (!vec_PIs.size())
		return true;
	
	//____________________________________________________________________________________
	// 		Write list of obsolete PIs to postgreSQL
	//____________________________________________________________________________________
	string sql_cmd (	"CREATE TEMP TABLE t__orphans (pi OID)");
	
	if (!pg_db.ExecCommandOk(sql_cmd.c_str()))
		return pg_print_sql_err(pg_db, table_name,
								"create a temporary table of entries marked for removal ",
								sql_cmd, CERR);
	
	if (! pg_export_PIs_to_table(	pg_db,
									"t__orphans",
									vec_PIs,
									VERBOSE, CERR))
		return false;
	
	//____________________________________________________________________________________
	// 		prepare SQL
	//____________________________________________________________________________________
	vector<string> descriptions;
	descriptions.push_back("create temporary index on the table of orphaned entries");
	if (archived_columns.length())
		descriptions.push_back("archive before deleting entries marked for removal");
	descriptions.push_back("remove entries marked for removal");
	descriptions.push_back("remove the temporary table of entries marked for removal");

	
	vector<string> sql_cmds;
	sql_cmds.push_back("CREATE UNIQUE INDEX t__i_orphans ON t__orphans(pi)");
	if (archived_columns.length())
		sql_cmds.push_back(
#ifndef NDEBUG
							"EXPLAIN ANALYSE "
#endif
							"INSERT INTO archived_" + table_name +
							" SELECT " + archived_columns +
							" FROM " + table_name + " NATURAL JOIN t__orphans");
	sql_cmds.push_back(
#ifndef NDEBUG
							"EXPLAIN ANALYSE "
#endif
						"DELETE FROM " + table_name +
						" WHERE t__orphans.pi = " + table_name + ".pi ");
	sql_cmds.push_back("DROP TABLE t__orphans");
	
	assert(sql_cmds.size() == descriptions.size());

	//____________________________________________________________________________________
	// 		postgreSQL removes all orphans
	//____________________________________________________________________________________
	for (unsigned i = 0; i < sql_cmds.size(); ++i)
	{
		ExecStatusType stat = pg_db.Exec(sql_cmds[i].c_str());
		if (stat !=  PGRES_COMMAND_OK && stat != PGRES_TUPLES_OK)
			return pg_print_sql_err(pg_db, table_name, descriptions[i], sql_cmds[i], CERR);
		if (stat == PGRES_TUPLES_OK)
		{
			int curr_batch_size = pg_db.Tuples();
			for (int i = 0; i < curr_batch_size; ++i)
				VERBOSE << "\n" << pg_db.GetValue(i, 0);
			VERBOSE << "\n";

		}
		VERBOSE << "..";

	}
	return true;
}



//________________________________________________________________________________________
// 		get current sequence value from postgres
//________________________________________________________________________________________
bool get_pgsql_sequence_currval(	t_pg_cursor&	pg_db,
									const string&	sequence_name,
									t_id&			current_value,
									ostream& 		CERR)
{
	string sql_cmd = "SELECT nextval('" + sequence_name + "')";
	if (!pg_db.ExecTuplesOk(sql_cmd.c_str()))
	{
		CERR		<< "\nERROR!! PostgreSQL error.\n\t"
					<< "Could not retreive the current value of the sequence "
					<< sequence_name
					<< "\n["
					<< sql_cmd << "]\n\t"
					<< pg_db.ErrorMessage();
		return false;
	}
	current_value = atol(pg_db.GetValue(0, 0)) - 1;
	return true;
}
bool set_pgsql_sequence_currval(	t_pg_cursor&	pg_db,
									const string&	sequence_name,
									const t_id&		new_value,
									ostream& 		CERR)
{
	string sql_cmd = "SELECT setval('" + sequence_name + "', ";
						ultoa(new_value, sql_cmd) += ")";
	if (!pg_db.ExecTuplesOk(sql_cmd.c_str()))
	{
		CERR		<< "\nERROR!! PostgreSQL error.\n\t"
					<< "Could not set the new value of the sequence "
					<< sequence_name
					<< "\n["
					<< sql_cmd << "]\n\t"
					<< pg_db.ErrorMessage();
		return false;
	}
	return true;
}

